package org.example.slave.channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.SneakyThrows;

import java.net.SocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @author dingrui
 * @since 2022/2/16
 * @description
 */
public abstract class MyChannelPool {

    // netty channel和自定义channel缓存
    private static final Map<Channel, MyChannel> chMgr = new ConcurrentHashMap<>();

    private static EventLoopGroup group = new NioEventLoopGroup();                              // 非阻塞IO线程组
    private static Bootstrap b = new Bootstrap();

    static {
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new BizHandler());
                    }
                });
    }

    @SneakyThrows
    public static MyChannel open(SocketAddress addr) {
        MyChannel myChannel = null;
        ChannelFuture f = b.connect(addr).sync();
        if (f.isSuccess()) {
            f.channel().pipeline().get(BizHandler.class).latch.await();
            // 获取到自定义channel实例
            myChannel = chMgr.get(f.channel());
        } else throw new RuntimeException(f.cause());
        if (Objects.isNull(myChannel)) throw new NullPointerException("myChannel");
        return myChannel;
    }


    /**
     * @author dingrui
     * @since 2022/2/16
     * @description 在客户端与mysql服务端通信过程中需要持续收发消息 采用handler职责链比较难控制 将channel实例暴露出去 将消息缓存 操纵内存缓存
     */
    public static class BizHandler extends SimpleChannelInboundHandler<ByteBuf> {

        private NettyChannel channel;
        private final CountDownLatch latch = new CountDownLatch(1);

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            // 连接上即将channel实例暴露出去
            this.channel = new NettyChannel(channel);
            // 缓存
            chMgr.put(channel, this.channel);
            this.latch.countDown();
            super.channelActive(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            this.channel.setCh(null);
            chMgr.remove(ctx.channel());
            super.channelInactive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
            if (Objects.isNull(this.channel)) return;
            // 接收的数据全部先写缓存
            this.channel.writeCache(byteBuf);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
}
